package com.lzx.spark.sanbao.tcsl.app;

import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

import static org.apache.spark.sql.functions.*;

/**
 * @author lzx
 * @version 1.0
 * @date 2024/8/24 下午4:38
 * @description: TODO
 */
public class ods_cater_item_sales_plan_full {
    public static void main(String[] args) {

        System.setProperty("HADOOP_USER_NAME", "hadoop");
        // 1 创建spark配置
        SparkConf conf = new SparkConf().setAppName("SparkSql");

        SparkSession spark = SparkSession.builder().enableHiveSupport().config(conf).getOrCreate();

//        String yesterday = LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd"));

        String yesterday = "20240822";


        // 定义 JSON Schema
        StructType itemListSchema = new StructType()
                .add("sizeId", DataTypes.LongType)
                .add("itemId", DataTypes.LongType)
                .add("itemName", DataTypes.StringType)
                .add("plId", DataTypes.LongType)
                .add("unitName", DataTypes.StringType)
                .add("sizeName", DataTypes.StringType)
                .add("memberPrice", DataTypes.DoubleType)
                .add("price", DataTypes.DoubleType)
                .add("itemCode", DataTypes.StringType)
                .add("costPrice", DataTypes.DoubleType)
                .add("unitId", DataTypes.LongType)
                .add("stdPrice", DataTypes.DoubleType);

        StructType shopDataSchema = new StructType()
                .add("enablePriLocalPriceFlg", DataTypes.BooleanType)
                .add("code", DataTypes.StringType)
                .add("belongShopId", DataTypes.LongType)
                .add("plShiftsMap", DataTypes.StringType) // 需要进一步解析
                .add("isMultiPricePlan", DataTypes.BooleanType)
                .add("plAreasMap", DataTypes.StringType) // 需要进一步解析
                .add("isDayLimit", DataTypes.BooleanType)
                .add("belongShopName", DataTypes.StringType)
                .add("isEnable", DataTypes.BooleanType)
                .add("delflg", DataTypes.IntegerType)
                .add("createShopId", DataTypes.LongType)
                .add("createShopName", DataTypes.StringType)
                .add("id", DataTypes.LongType)
                .add("brandCode", DataTypes.StringType)
                .add("brandName", DataTypes.StringType)
                .add("saleTypeId", DataTypes.IntegerType)
                .add("isWeekLimit", DataTypes.BooleanType)
                .add("saleTypeName", DataTypes.StringType)
                .add("isO2oPricePlan", DataTypes.BooleanType)
                .add("sortOrder", DataTypes.IntegerType)
                .add("plWeeksMap", DataTypes.StringType) // 需要进一步解析
                .add("brandId", DataTypes.LongType)
                .add("name", DataTypes.StringType)
                .add("isTimeLimit", DataTypes.BooleanType)
                .add("itemList", DataTypes.StringType) // 需要进一步解析
                .add("describe", DataTypes.StringType)
                .add("isShiftLimit", DataTypes.BooleanType)
                .add("priId", DataTypes.LongType)
                .add("plPosMap", DataTypes.StringType); // 需要进一步解析

        // 查询获取 shopid 和 shopdata
        Dataset<Row> showResult = spark.sql("SELECT shopid, shopdata FROM sanbao.ods_cater_item_sales_plan_top1_full WHERE dt = '" + yesterday + "'");

        // 去除头尾的 [] 符号并解析 JSON 数据
        Dataset<Row> parsedData = showResult
                .withColumn("cleanedShopData", expr("substring(shopdata, 2, length(shopdata) - 2)"))
                .withColumn("parsedShopData", from_json(col("cleanedShopData"), DataTypes.createArrayType(shopDataSchema)))
                .withColumn("parsedShopData", explode(col("parsedShopData")))
                .select(col("shopid"), col("parsedShopData.*"));

        parsedData.show(false);

//        // 解析内嵌的 JSON 字符串字段
//        Dataset<Row> itemListParsed = parsedData
//                .withColumn("itemList", from_json(col("itemList"), DataTypes.createArrayType(itemListSchema)))
//                .withColumn("itemList", explode(col("itemList")))
//                .select(
//                        col("shopid"),
//                        col("itemList.sizeId"),
//                        col("itemList.itemId"),
//                        col("itemList.itemName"),
//                        col("itemList.plId"),
//                        col("itemList.unitName"),
//                        col("itemList.sizeName"),
//                        col("itemList.memberPrice"),
//                        col("itemList.price"),
//                        col("itemList.itemCode"),
//                        col("itemList.costPrice"),
//                        col("itemList.unitId"),
//                        col("itemList.stdPrice")
//                );

        parsedData = parsedData.withColumn("dt", lit(yesterday));

        parsedData.write()
                .mode("overwrite") // 使用 append 模式插入数据
                .insertInto("sanbao.dim_cater_item_sales_plan_full");

//        itemListParsed.show(false);

        spark.stop();


            }
 }


